package io.kiki.stack.netty.protocol;


import io.kiki.stack.netty.command.*;
import io.kiki.stack.netty.config.switches.ProtocolSwitch;
import io.kiki.stack.netty.invoker.RpcCommandType;
import io.kiki.stack.netty.util.CrcUtil;
import io.kiki.stack.netty.util.ThreadLocalArriveTimeHolder;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.List;

@Slf4j
@Setter
@Getter
public class DecoderImpl implements Decoder {


    private final int lessLen;

    {
        lessLen = Math.min(ProtocolImpl.getResponseHeaderLength(), ProtocolImpl.getRequestHeaderLength());
    }


    @Override
    public void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> out) throws Exception {
        // the less length between response header and request header
        if (in.readableBytes() >= lessLen) {
            in.markReaderIndex();
            byte protocol = in.readByte();
            in.resetReaderIndex();

            /*
             * ver: version for protocol
             * type: request/response/request oneway
             * cmdcode: code for remoting command
             * ver2:version for remoting command
             * requestId: id of request
             * codec: code for codec
             * switch: function switch
             * (req)timeout: request timeout
             * (resp)respStatus: response status
             * classLen: length of request or response class name
             * headerLen: length of header
             * contentLen: length of content
             * className
             * header
             * content
             */
            if (in.readableBytes() > 2 + 1) {
                int startIndex = in.readerIndex();
                in.markReaderIndex();
                in.readByte(); //protocol code
                //protocol version
                byte type = in.readByte(); //type
                if (type == RpcCommandType.REQUEST || type == RpcCommandType.REQUEST_ONEWAY) {
                    //decode request
                    if (in.readableBytes() >= ProtocolImpl.getRequestHeaderLength() - 3) {
                        short cmdCode = in.readShort();
                        int requestId = in.readInt();
                        byte protocolSwitchValue = in.readByte();
                        int timeout = in.readInt();
                        short classLen = in.readShort();
                        short headerLen = in.readShort();
                        int contentLen = in.readInt();
                        byte[] clazz = null;
                        byte[] header = null;
                        byte[] content = null;

                        Channel channel = channelHandlerContext.channel();
                        ThreadLocalArriveTimeHolder.arrive(channel, requestId);

                        // decide the at-least bytes length for each version

                        boolean crcSwitchOn = ProtocolSwitch.isOn(protocolSwitchValue, ProtocolSwitch.CRC_SWITCH_INDEX);
                        int lengthAtLeastForV2 = classLen + headerLen + contentLen;
                        if (crcSwitchOn) {
                            lengthAtLeastForV2 += 4;// crc int
                        }

                        // continue read
                        if ((in.readableBytes() >= lengthAtLeastForV2)) {
                            if (classLen > 0) {
                                clazz = new byte[classLen];
                                in.readBytes(clazz);
                            }
                            if (headerLen > 0) {
                                header = new byte[headerLen];
                                in.readBytes(header);
                            }
                            if (contentLen > 0) {
                                content = new byte[contentLen];
                                in.readBytes(content);
                            }
                            if (crcSwitchOn) {
                                checkCRC(in, startIndex);
                            }
                        } else {// not enough data
                            in.resetReaderIndex();
                            return;
                        }

                        long headerArriveTimeInNano = ThreadLocalArriveTimeHolder.getAndClear(channel, requestId);

                        RequestCommand command;
                        if (cmdCode == CommandCode.heartbeat) {
                            command = new HeartbeatRequestCommand();
                        } else {
                            command = createRequestCommand(cmdCode, headerArriveTimeInNano);
                        }
                        command.setType(type);
                        //command.setVersion(ver2);
                        command.setId(requestId);
                        //command.setSerializer(serializer);
                        command.setProtocolSwitch(ProtocolSwitch.create(protocolSwitchValue));
                        command.setTimeout(timeout);
                        command.setClazz(clazz);
                        command.setHeader(header);
                        command.setContent(content);

                        out.add(command);
                    } else {
                        in.resetReaderIndex();
                    }
                } else
                    if (type == RpcCommandType.RESPONSE) {
                        //decode response
                        if (in.readableBytes() >= ProtocolImpl.getResponseHeaderLength() - 3) {
                            short cmdCode = in.readShort();
                            int requestId = in.readInt();
                            byte protocolSwitchValue = in.readByte();
                            short status = in.readShort();
                            short classLen = in.readShort();
                            short headerLen = in.readShort();
                            int contentLen = in.readInt();
                            byte[] clazz = null;
                            byte[] header = null;
                            byte[] content = null;

                            // decide the at-least bytes length for each version
                            boolean crcSwitchOn = ProtocolSwitch.isOn(protocolSwitchValue, ProtocolSwitch.CRC_SWITCH_INDEX);
                            int lengthAtLeastForV2 = classLen + headerLen + contentLen;
                            if (crcSwitchOn) {
                                lengthAtLeastForV2 += 4;// crc int
                            }

                            // continue read
                            if ((in.readableBytes() >= lengthAtLeastForV2)) {
                                if (classLen > 0) {
                                    clazz = new byte[classLen];
                                    in.readBytes(clazz);
                                }
                                if (headerLen > 0) {
                                    header = new byte[headerLen];
                                    in.readBytes(header);
                                }
                                if (contentLen > 0) {
                                    content = new byte[contentLen];
                                    in.readBytes(content);
                                }
                                if (crcSwitchOn) {
                                    checkCRC(in, startIndex);
                                }
                            } else {// not enough data
                                in.resetReaderIndex();
                                return;
                            }
                            ResponseCommand command;
                            if (cmdCode == CommandCode.heartbeat) {
                                command = new HeartbeatResponseCommand();
                            } else {
                                command = createResponseCommand(cmdCode);
                            }
                            command.setType(type);
                            //command.setVersion(ver2);
                            command.setId(requestId);
                            //command.setSerializer(serializer);
                            command.setProtocolSwitch(ProtocolSwitch.create(protocolSwitchValue));
                            command.setResponseStatus(ResponseType.valueOf(status));
                            command.setClazz(clazz);
                            command.setHeader(header);
                            command.setContent(content);
                            command.setResponseTimeMillis(System.currentTimeMillis());
                            command.setResponseHost((InetSocketAddress) channelHandlerContext.channel().remoteAddress());

                            out.add(command);
                        } else {
                            in.resetReaderIndex();
                        }
                    } else {
                        String emsg = "Unknown command type: " + type;
                        log.error(emsg);
                        throw new RuntimeException(emsg);
                    }
            }


        }
    }

    private void checkCRC(ByteBuf byteBuf, int startIndex) {
        int endIndex = byteBuf.readerIndex();
        int expectedCrc = byteBuf.readInt();
        byte[] frame = new byte[endIndex - startIndex];
        byteBuf.getBytes(startIndex, frame, 0, endIndex - startIndex);
        int actualCrc = CrcUtil.crc32(frame);
        if (expectedCrc != actualCrc) {
            String err = "CRC check failed!";
            log.error(err);
            throw new RuntimeException(err);
        }
    }

    private ResponseCommand createResponseCommand(short cmdCode) {
        ResponseCommand responseCommand = new AbstractResponseCommand();
        responseCommand.setCommandCode(CommonCommandCode.valueOf(cmdCode));
        return responseCommand;
    }

    private AbstractRequestCommand createRequestCommand(short cmdCode, long headerArriveTimeInNano) {
        AbstractRequestCommand requestCommand = new AbstractRequestCommand();
        requestCommand.setCommandCode(CommonCommandCode.valueOf(cmdCode));
        requestCommand.setArriveTime(System.currentTimeMillis());
        requestCommand.setArriveHeaderTimeInNano(headerArriveTimeInNano);
        requestCommand.setArriveBodyTimeInNano(System.nanoTime());
        return requestCommand;
    }

}
